package com.sfx.concurrent.demo.concurrent.BlockingQueue;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


@Slf4j(topic = "c.MyBlockingQueue")
public class MyBlockingQueue<T> {
    //队列
    private Deque<T> queue =new ArrayDeque<T>();
    //加锁
    private ReentrantLock lock = new ReentrantLock();
    //阻塞队列大小
    private int capacity;
    //阻塞添加休息室
    Condition fullWaitSet = lock.newCondition();
    //阻塞获取休息室
    Condition emptyWaitSet = lock.newCondition();

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    //增加超时获取
    public T poll(long timeout, TimeUnit unit) {
        lock.lock();
        try{
            //将timeout转化为nanos
            long nanos = unit.toNanos(timeout);
            while(queue.isEmpty()) {
                if(nanos<=0) {
                    break;
                }
                try {
                    /**
                     * 该方法返回剩余等待的时间
                     */
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally{
            lock.unlock();
        }
    }

    //阻塞获取
    public T take() {
        lock.lock();
        try{
            while(queue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally{
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T element) {
        lock.lock();
        try{
            while(queue.size()==this.capacity) {
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //获取阻塞队列大小
    public int size() {
        lock.lock();
        try{
            return this.queue.size();
        }finally {
            lock.unlock();
        }
    }
}